package link.jfire.disqueue;

import link.jfire.disqueue.util.SequenceUtil;

public class SingleProducer<T> implements Producer
{
    private long         next                   = Sequencer.InitValue + 1;
    private long         cacheGate              = Sequencer.InitValue;
    private int          ringBufferSize;
    private Sequencer    ringBufferSequencer;
    private Sequencer[]  entryHandlerSequencers = new Sequencer[0];
    private ProducerWaitStrategy waitStrategy;
    
    SingleProducer(RingBuffer<T> ringBuffer)
    {
        ringBufferSize = ringBuffer.size();
        ringBufferSequencer = ringBuffer.getSequencer();
        if (ringBuffer.getEntryHandlerSequencers().length == 0)
        {
            entryHandlerSequencers = new Sequencer[] { ringBufferSequencer };
        }
        else
        {
            entryHandlerSequencers = ringBuffer.getEntryHandlerSequencers();
        }
    }
    
    @Override
    public long next()
    {
        if (next - cacheGate == ringBufferSize)
        {
            cacheGate =SequenceUtil.min(entryHandlerSequencers);
        }
        long result = next;
        next += 1;
        return result;
    }
    
    @Override
    public void publish(long sequence)
    {
        ringBufferSequencer.directSet(sequence);
        waitStrategy.signalAllWait();
    }
    
}
